package blue.mqtt4j.internal.client.net;

import blue.mqtt4j.client.MqttClientException;
import blue.mqtt4j.client.MqttOptions;
import blue.mqtt4j.client.MqttQoS;
import blue.mqtt4j.client.MqttTopic;
import blue.mqtt4j.core.handler.SenderListener;
import blue.mqtt4j.core.util.MessageIdGen;
import blue.mqtt4j.core.util.MqttMessageUtil;
import blue.mqtt4j.internal.client.impl.DefaultMqttClient;
import blue.mqtt4j.internal.client.impl.PublishHandler;
import blue.mqtt4j.internal.client.task.ScheduledTaskExecutor;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageBuilders;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
import io.netty.handler.codec.mqtt.MqttUnsubscribeMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.Collection;

/**
 * @author Jin Zheng
 * @since 1.0 2020-09-30
 */
public class NettyClient
{
	private static Logger logger = LoggerFactory.getLogger(NettyClient.class);

	private final InetSocketAddress remoteAddress;
	private final MqttOptions options;
	private final DefaultMqttClient mqttClient;
	private final ScheduledTaskExecutor taskExecutor;
	private final PublishHandler publishHandler;
	private final MqttInitializer initializer;
	private final EventLoopGroup workerGroup;
	private final Bootstrap bootstrap;
	private Channel channel;
	private boolean stop = false;

	public NettyClient(String broker, MqttOptions options, DefaultMqttClient mqttClient)
	{
		this.remoteAddress = this.getRemoteAddress(broker);
		this.options = options;
		this.mqttClient = mqttClient;
		this.taskExecutor = new ScheduledTaskExecutor();
		this.publishHandler = new PublishHandler(options.timeoutMillis());
		this.initializer = new MqttInitializer(this, options);
		this.workerGroup = new NioEventLoopGroup(2);
		bootstrap = new Bootstrap();
		bootstrap.group(workerGroup)
				.channel(NioSocketChannel.class)
				.handler(initializer);
	}

	public void publish(String topic, MqttQoS qos, byte[] payload, ChannelFutureListener listener)
	{
		ByteBuf buf = Unpooled.copiedBuffer(payload);
		MqttPublishMessage message = MqttMessageBuilders.publish()
				.messageId(MessageIdGen.getInstance().id())
				.topicName(topic)
				.qos(qos.toNettyMqttQoS())
				.payload(buf)
				.build();
		publishHandler.publish(message, listener);
	}

	public void receive(MqttPublishMessage message)
	{
		mqttClient.receive(message);
	}

	public int subscribe(Collection<MqttTopic> topicList)
	{
		if (topicList == null || topicList.isEmpty())
			return -1;

		int messageId = MessageIdGen.getInstance().id();
		MqttMessageBuilders.SubscribeBuilder builder = MqttMessageBuilders.subscribe();
		for (MqttTopic topic : topicList)
		{
			builder.addSubscription(topic.qos().toNettyMqttQoS(), topic.topic());
		}
		MqttSubscribeMessage message = builder.messageId(messageId).build();
		taskExecutor.schedule(messageId, message);
		return messageId;
	}

	public void subAck(int messageId)
	{
		taskExecutor.cancel(messageId);
	}

	public int unsubscribe(String topic)
	{
		int messageId = MessageIdGen.getInstance().id();
		MqttUnsubscribeMessage message = MqttMessageBuilders.unsubscribe()
				.addTopicFilter(topic)
				.messageId(messageId)
				.build();
		taskExecutor.schedule(messageId, message);
		return messageId;
	}

	public void unsubAck(int messageId)
	{
		taskExecutor.cancel(messageId);
		mqttClient.unsubAck(messageId);
	}

	public void pubRec(int messageId)
	{
		publishHandler.setPubRec(messageId);
		MqttMessage message = MqttMessageUtil.createPubRel(messageId);
		publishHandler.publish(message, null);
	}

	public void pubRel(int messageId)
	{
		mqttClient.pubRel(messageId);
	}

	public void sendPubAck(int messageId)
	{
		MqttMessage message = MqttMessageBuilders.pubAck()
				.packetId((short)messageId)
				.reasonCode((byte)0)
				.build();
		channel.writeAndFlush(message).addListener(new SenderListener(message));
	}

	public void sendPubRec(int messageId)
	{
		MqttMessage message = MqttMessageUtil.createPubRec(messageId);
		channel.writeAndFlush(message).addListener(new SenderListener(message));
	}

	public void sendPubComp(int messageId)
	{
		MqttMessage message = MqttMessageUtil.createPubComp(messageId);
		publishHandler.publish(message, null);
	}

	public void connect()
	{
		this.doConnect();
	}

	public void doConnect()
	{
		synchronized (bootstrap)
		{
			bootstrap.connect(remoteAddress).addListener(new ChannelFutureListener()
			{
				@Override
				public void operationComplete(ChannelFuture future) throws Exception
				{
					if (!future.isSuccess())
					{
						future.channel().pipeline().fireChannelInactive();
						return;
					}
					logger.debug("Connected successful: {}", remoteAddress);
				}
			});
		}
	}

	public void onDisconnected()
	{
		channel = null;
		taskExecutor.setChannel(null);
		//publishHandler.setChannel(null);
		mqttClient.onDisconnected();
	}

	public void onConnected(Channel channel, boolean sessionPresent)
	{
		this.channel = channel;
		taskExecutor.setChannel(channel);
		publishHandler.setChannel(channel);
		mqttClient.onConnected(sessionPresent);
	}

	public void disconnect()
	{
		stop = true;
		publishHandler.stop();
		if (channel != null)
		{
			MqttMessage message = MqttMessageBuilders.disconnect().reasonCode((byte)0).build();
			channel.writeAndFlush(message);
			channel.flush().close().syncUninterruptibly();
			logger.debug("Channel closed.");
		}
		if (workerGroup != null)
		{
			workerGroup.shutdownGracefully();
			logger.debug("EventLoopGroup closed.");
		}
		logger.debug("Disconnected successful: {}", remoteAddress);
	}

	public boolean isStop()
	{
		return stop;
	}

	public InetSocketAddress getRemoteAddress()
	{
		return remoteAddress;
	}

	public PublishHandler getPublishHandler()
	{
		return publishHandler;
	}

	private InetSocketAddress getRemoteAddress(String broker)
	{
		String[] addrs = broker.split(":");
		if (addrs.length != 2)
			throw new MqttClientException("Invalid address: " + broker);

		return new InetSocketAddress(addrs[0], Integer.parseInt(addrs[1]));
	}

}
